package mapReduce.demo06_sort.demo02_reverseIndex;

import mapReduce.util.JobSubmit;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class ReverseIndexApp {

    public static class RIMapper extends Mapper<LongWritable, Text,Text,Text>{

        Text outKey = new Text();
        Text outValue = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {

            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            //文件名充当平台名
            String platform = inputSplit.getPath().getName();
            outValue.set(platform);

            String line = value.toString();
            String[] products = line.split("\\s");

            //每个字符串就是一个商品
            for (String product : products) {
                outKey.set(product);
                context.write(outKey,outValue);
            }

        }
    }

    public static class RIReduer extends Reducer<Text,Text,Text,Text>{

        Text outValue = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {

            /**
             * 接收到的数据
             *      商品名 (平台,平台,....)
             */

            StringBuffer sb = new StringBuffer();

            for (Text value : values) {
                sb.append(value.toString()+",");
            }

            String line = sb.toString();
            line = line.replace(".txt","");
            line = line.substring(0,line.lastIndexOf(","));

            outValue.set(line);

            context.write(key,outValue);
        }
    }

    public static void main(String[] args) {

        JobSubmit.submitBaseJob(ReverseIndexApp.class,args);

    }

}
